package com.test
import com.test.GoodtoGood.spark
import org.apache.hadoop.hive.common.AcidMetaDataFile.DataFormat
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{DoubleType, IntegerType, StructType}
import org.apache.spark.sql.functions._
object Main {
  // 创建 SparkSession
  val spark = SparkSession.builder().appName("als-demo").master("local[*]").getOrCreate()

  // 修改mapUserIdAndGoodsRatings方法
  def mapUserIdAndGoodsRatingsPlus(): DataFrame = {

    val filePaths: List[String] = List("D:\\PycharmProjects\\wordpress\\active_ran_buy.txt",
      "D:\\PycharmProjects\\wordpress\\active_ran_collect.txt",
      "D:\\PycharmProjects\\wordpress\\active_ran_like.txt",
      "D:\\PycharmProjects\\wordpress\\active_ran_browse.txt")


    // 定义数据结构
    val schema = new StructType()
      .add("userId", IntegerType)
      .add("itemId", IntegerType)
      .add("rating", DoubleType)

    // 读取数据并应用定义的数据结构
    val allRatings = filePaths.map { filePath =>
      spark.read
        .schema(schema)
        .options(Map("inferSchema" -> "false", "delimiter" -> ", ", "header" -> "false"))
        .csv(filePath)
    }

    // 合并所有数据
    val combinedRatings = allRatings.reduce((df1, df2) => df1.union(df2))


    // 将DataFrame转换为RDD，指定所需的数据类型
    val rating_value: Double = 1.0
    val rddData: RDD[((Int, Int), Double)] = combinedRatings.rdd.map {
      case Row(userId: Int, itemId: Int, rating: Double) =>
        ((userId, itemId), rating_value)
    }
    // 对相同的(userid, goodid)的rating值进行相加
    val aggregatedRatings = rddData.reduceByKey(_ + _)
      .map { case ((userId, itemId), ratingSum) => (userId, (itemId, ratingSum)) }

    val aggregatedRatingsRow = aggregatedRatings.map {
      case (userId, (itemId, ratingSum)) =>
      Row(userId, itemId, ratingSum)
    }

    // 将RDD转换为DataFrame
    val dataFrame1 = spark.createDataFrame(aggregatedRatingsRow, schema)

//    dataFrame1.rdd.map(row => (row.getInt(0), row.getInt(1), row.getDouble(2)))

    dataFrame1
  }

  def UserGoodsRating():DataFrame = {
    // 定义数据结构
    val schema = new StructType()
      .add("userId", IntegerType)
      .add("itemId", IntegerType)
      .add("rating", DoubleType)

    // 读取数据时应用定义的数据结构
    val rating = spark.read
      .schema(schema)
      .options(Map("inferSchema" -> "false", "delimiter" -> ", ", "header" -> "true"))
      .csv("D:\\PycharmProjects\\wordpress\\active_ran_collect.txt")

    rating
  }

  def main(args: Array[String]): Unit = {
    // 设置日志级别
    Logger.getLogger("org").setLevel(Level.ERROR)


//    val rating = UserGoodsRating()
    val rating = mapUserIdAndGoodsRatingsPlus()


    // 展示前5条评分记录
    rating.show(20)
    // |10050|[-0.6172592, 0.6200862, -0.073221095, 1.2677336, -0.19729076, -0.71175224]   |
    // |10000|[0.60912734, -0.34276113, 0.034928102, -0.47109106, -0.5147739, -0.49815503] |
    // 创建 ALS 模型
    val als = new ALS()
      .setMaxIter(15) // 迭代次数，用于最小二乘交替迭代的次数
      .setRank(6) // 隐向量的维度
      .setRegParam(0.1) // 惩罚系数
      .setUserCol("userId") // user_id
      .setItemCol("itemId") // item_id
      .setRatingCol("rating") // 评分列

    // 训练模型
    val model = als.fit(rating)

    // 打印用户向量和物品向量
    model.userFactors.show(truncate = false)
    model.itemFactors.show(truncate = false)


    println("-------开始推荐商品---------")
    // 给所有用户推荐2个物品
//    model.recommendForAllUsers(2).show()
    model.recommendForAllUsers(3).show(truncate = false)

    // 给所有用户推荐20个物品
//    val recommendations = model.recommendForAllUsers(5).show()


    // 保存数据


    // 停止 SparkSession
    spark.stop()
  }
}
